package com.xzx.flink.streamapi.source;

import com.xzx.flink.bean.SensorReading;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.Random;

/**
 * 自定义source
 *
 * @author xinzhixuan
 * @version 1.0
 * @date 2021-08-24 10:59 下午
 */
public class SourceFromCustomSource {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<SensorReading> dataStream = env.addSource(new MySource()).setParallelism(2);

        dataStream.print("tt");

        env.execute(SourceFromCustomSource.class.getName());
    }

    static class MySource implements SourceFunction<SensorReading> {

        private boolean running = true;
        @Override
        public void run(SourceContext<SensorReading> ctx) throws Exception {
            while (running) {
                ctx.collect(new SensorReading("resources" + new Random().nextInt(10), System.currentTimeMillis(), 32.0 + new Random().nextDouble()));
                Thread.sleep(1000);
            }
        }

        @Override
        public void cancel() {
            running = false;
        }
    }
}

